package com.yonyou.findata.kafka.adminop;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import kafka.admin.TopicCommand;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 *
 * 0.11.0.0 之前, 可以使用kafak.admin.AdminClient 和 kafka.admin.AdminUtils
 * 0.11.0.0 之后, 使用org.apache.kafka.clients.admin.KafkaAdminClient
 *
 *
 * @author pizhihui
 * @date 2020-06-17
 */
public class AdminTest {


    public static void main(String[] args) throws ExecutionException, InterruptedException {

        // AdminOpUtils.createTopic("test-1", 4 ,1);
        //
        // AdminOpUtils.listTopic();
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);


//        NewTopic newTopic = new NewTopic("test-3", 4, (short) 1);
//        CreateTopicsResult res = client.createTopics(Lists.newArrayList(newTopic));
//        System.out.println(JSONObject.toJSONString(res.all().get()));


        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult listTopicsResult = client.listTopics(listTopicsOptions);
        System.out.println(listTopicsResult.names().get());
        System.out.println(listTopicsResult.namesToListings().get());
        System.out.println(JSONObject.toJSONString(listTopicsResult.listings().get()));



        client.close();
    }

}
